package nl.justobjects.pushlet.pushmanager;
import java.io.Serializable;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

import nl.justobjects.pushlet.core.Dispatcher;
import nl.justobjects.pushlet.core.Event;
import nl.justobjects.pushlet.core.EventSource;
import nl.justobjects.pushlet.core.Protocol;
import nl.justobjects.pushlet.util.Log;
import nl.justobjects.pushlet.util.Sys;
import nl.justobjects.pushlet.wrapper.JMSService;

/**
 * xnkl  14-3-10.
 */
public class PushThread implements EventSource, Runnable {
    private volatile boolean alive = true;
    private volatile boolean active = true;
    private long sleepTime = 0;
    private long delay = 0;
    private String eventName;
    private String subject;
    private String label;
    private Map<String, Serializable> data = new HashMap<String, Serializable>();
    private Map<String, Object> createDataParams = null;
    private Map<String, Object> whenFinishParams = null;
    private long endTime = -1l;
    private JMSService jmsService = null;
    private DataCreator dataCreator = null;
    private WhenFinish whenFinish = null;
    private String threadId;
    private List<String> pids;
    
    public String getThreadId() {
		return threadId;
	}
	public String getSubject() {
		return subject;
	}
	public String getLabel() {
		return label;
	}
    public long getEndTime() {
		return endTime;
	}
    public boolean isAlive() {
        return alive;
    }
    public boolean isActive() {
        return active;
    }

    public PushThread(String threadId, String subject, String label) {
        this.subject = subject;
        this.label = label;
        this.eventName = subject + "_" + label;
        this.threadId = threadId;
    }
    
    /**
     * 主题内多播，指定pid
     * @param pids
     * @return
     */
    public PushThread setSubscribers(List<String> pids) {
    	this.pids = pids;
    	return this;
    }

    /**
     * 设置事件源推送时间间隔和存活时间（毫秒）
     * @param interval
     * @param runTime
     * @return
     */
    public PushThread setThreadLife(long interval, long runTime) {
        this.sleepTime = interval;
        this.endTime = runTime + Sys.now();
        return this;
    }

    /**
     * 设置推送数据的创建者
     * @param dataCreator
     * @param createDataParams
     * @return
     */
    public PushThread setDataCreator(DataCreator dataCreator, Map<String, Object> createDataParams) {
        this.dataCreator = dataCreator;
        this.createDataParams = createDataParams;
        return this;
    }

    /**
     * 设置推送结束后执行函数
     * @param whenFinish
     * @param whenFinishParams
     * @return
     */
    public PushThread setWhenFinish(WhenFinish whenFinish, Map<String, Object> whenFinishParams) {
        this.whenFinish = whenFinish;
        this.whenFinishParams = whenFinishParams;
        return this;
    }

    /**
     * 设置推送的数据
     * @param data
     * @return
     */
    public PushThread setPushData(Map<String, ? extends Serializable> data) {
        this.data = (Map<String, Serializable>) data;
        return this;
    }

    /**
     * 设置JMS服务，如果null，则只能本机推送
     * @param jmsService
     * @return
     */
    public PushThread setJMSService(JMSService jmsService) {
        this.jmsService = jmsService;
        return this;
    }

    /**
     * 设置延时
     * @param delay
     * @return
     */
    public PushThread setDelay(long delay) {
        this.delay = delay;
        return this;
    }

    public Event pullEvent() {
        Event event = Event.createDataEvent(subject);
        if (label != null)
        	event.setField(Protocol.P_LABEL, label);
        if (dataCreator != null) {
        	// 设定了dataCreator，则每次通过dataCreator产生需要推送的数据
            Map<String, Serializable> createdData = (Map<String, Serializable>) dataCreator.createPushData(createDataParams);
            if (createdData != null)
        	    event.setFields(createdData);
        } else {
        	// 否则每次都推送同一份数据
	        if (data != null)
	            event.setFields(data);
        }
        return event;
    }

    @Override
    public void stop() {
        if (!alive)
            return;
        alive = false;
        if (!active) {
        	synchronized (this) {
				active = true;
				notifyAll();
			}
        }
    }

    @Override
    public synchronized void activate() {
        if (active || !alive)
            return;
        active = true;
        Log.debug("PushThread["+eventName+ "] activated.");
        notifyAll();
    }

    @Override
    public void passivate() {
        if (!active || !alive)
            return;
        active = false;
        Log.debug("PushThread["+eventName+ "] passivated.");
    }

    public void run() {
    	Log.debug("PushThread["+eventName+"] start.");
        // 延时
        if (delay > 0) {
            try {
                Thread.sleep(delay);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        while (alive) {
            // 推送数据
            try {
                Event event = pullEvent();
                if (jmsService != null) {
                	if (pids != null && pids.size() > 0) {
                		// 指定了订阅者的id，采取多播方式推送消息
                		jmsService.multicastEvent(pids, event);
                	} else {
                		// 没有指定，则广播推送，订阅了该主题的所有订阅者都将收到消息
                		jmsService.broadcastEvent(event);
                	}
                } else {
                	// 没有JMS服务只在本机域内推送
                	if (pids != null && pids.size() > 0) {
                		for (String pid : pids) {
                			Dispatcher.getInstance().unicast(event, pid);
                		}
                	} else {
                		Dispatcher.getInstance().multicast(event);
                	}
                }
            } catch (Throwable t) {
                Log.warn("EventPullSource exception while multicasting ", t);
                t.printStackTrace();
            }
            // 间隔时间
            try {
                if (sleepTime > 0)
                    Thread.sleep(sleepTime);
                if (!alive)
                    break;
                synchronized (this) {
                    while (!active) {
                        Log.debug("PushThread[" + eventName + "] waiting.");
                        wait();
                    }
                }
            } catch (InterruptedException e) {
                break;
            }
            // 检查是否过期
            if (Sys.now() >= endTime) {
                alive = false;
                break;
            }
        }
        // 线程结束回调函数
        if (whenFinish != null)
        	whenFinish.lastEvent(whenFinishParams);
        Log.debug("PushThread["+eventName+"] stop.");
    }
}
